package org.lzfto.flink.demo.business.flinkManager;

import java.io.IOException;
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.Paths;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import org.apache.commons.lang3.StringUtils;
import org.lzfto.flink.demo.common.result.PubResult;
import org.lzfto.flink.demo.config.MyAppProperties;
import org.springframework.core.io.FileSystemResource;
import org.springframework.http.MediaType;
import org.springframework.stereotype.Service;
import org.springframework.web.multipart.MultipartFile;
import org.springframework.web.reactive.function.BodyInserters;
import org.springframework.web.reactive.function.client.WebClient;
import org.springframework.web.reactive.function.client.WebClientResponseException;

import reactor.core.publisher.Mono;

/**
*
*@author lzfto
*@创建时间 2025年7月1日 14:24:59
*/
@Service
public class FlinkManageService {

	private final MyAppProperties prop;

	public FlinkManageService(MyAppProperties appProperties) {
		this.prop=appProperties;
	}

    public PubResult<Boolean> cancelJob(String jobId,String serverUrl) {
	    String url = (StringUtils.isNotEmpty(serverUrl)?serverUrl:this.prop.getFlinkAddress()) + "jobs/" + jobId;
	    //使用webclient的PATCH方法处理.
	    //如果成功，返回的字符串={}
	    try {
	        WebClient webClient = WebClient.create();
	        String response = webClient.patch()
	            .uri(url)
	            .retrieve()
	            .onStatus(status -> !status.is2xxSuccessful(),
	                clientResponse -> Mono.error(new RuntimeException("Failed to cancel job: " + clientResponse.statusCode())))
	            .bodyToMono(String.class)
	            .block(); // 阻塞等待结果，保持与原有接口兼容
	        if (response.equals("{}"))
	            return PubResult.ok_data(true);
	        else
	            return PubResult.error("取消失败:" + response);
	    } catch (WebClientResponseException e) {
	        return PubResult.error("Error cancelling job: " + e.getStatusCode() + " - " + e.getMessage());
	    } catch (Exception e) {
	        return PubResult.error("Error connecting to Flink server: " + e.getMessage());
	    }
	}

	public PubResult<Boolean> deleteJar(String jarId,String serverUrl) {
	    String url = (StringUtils.isNotEmpty(serverUrl)?serverUrl:this.prop.getFlinkAddress()) + "jars/" + jarId;
	    //如果成功，返回的字符串={}
	    try {
	        WebClient webClient = WebClient.create();
	        String response = webClient.delete()
	                .uri(url)
	                .retrieve()
	                .onStatus(status -> !status.is2xxSuccessful(),
	                        clientResponse -> Mono.error(new RuntimeException("Failed to cancel job: " + clientResponse.statusCode())))
	                .bodyToMono(String.class)
	                .block(); // 阻塞等待结果，保持与原有接口兼容
	        if (response.equals("{}"))
	            return PubResult.ok_data(true);
	        else
	            return PubResult.error("删除失败:" + response);
	    } catch (WebClientResponseException e) {
	        return PubResult.error("Error cancelling job: " + e.getStatusCode() + " - " + e.getMessage());
	    } catch (Exception e) {
	        return PubResult.error("Error connecting to Flink server: " + e.getMessage());
	    }
	}

	public PubResult<String> getJars(String serverUrl) {
	    try {
	        // 使用WebClient发起响应式HTTP请求
	        WebClient webClient = WebClient.create();
	        String url = (StringUtils.isNotEmpty(serverUrl)?serverUrl:this.prop.getFlinkAddress()) + "jars";
	        // 发送GET请求并异步处理响应
	        String response = webClient.get()
	                .uri(url)
	                .retrieve()
	                .onStatus(status -> !status.is2xxSuccessful(),
	                        clientResponse -> Mono.error(new RuntimeException("Failed to get jobs from Flink server: " + clientResponse.statusCode())))
	                .bodyToMono(String.class)
	                .block(); // 阻塞等待结果，保持与原有接口兼容
	
	        return PubResult.ok_data(response);
	    } catch (WebClientResponseException e) {
	        return PubResult.error("Error response from Flink server: " + e.getStatusCode() + " - " + e.getMessage());
	    } catch (Exception e) {
	        return PubResult.error("Error connecting to Flink server: " + e.getMessage());
	    }
	}

	public PubResult<String> getJobs(String serverUrl) {
        try {
            // 使用WebClient发起响应式HTTP请求
            WebClient webClient = WebClient.create();
            String url = (StringUtils.isNotEmpty(serverUrl)?serverUrl:this.prop.getFlinkAddress()) + "jobs/overview";
            // 发送GET请求并异步处理响应
            String response = webClient.get()
                .uri(url)
                .retrieve()
                .onStatus(status -> !status.is2xxSuccessful(),
                    clientResponse -> Mono.error(new RuntimeException("Failed to get jobs from Flink server: " + clientResponse.statusCode())))
                .bodyToMono(String.class)
                .block(); // 阻塞等待结果，保持与原有接口兼容

            return PubResult.ok_data(response);
        } catch (WebClientResponseException e) {
            return PubResult.error("Error response from Flink server: " + e.getStatusCode() + " - " + e.getMessage());
        } catch (Exception e) {
            return PubResult.error("Error connecting to Flink server: " + e.getMessage());
        }
    }

	/**
	 * 运行jar程序
	 * @param param 参数，非空
	 */
    public PubResult<String> runJar(JarJobParam param) {
	    String url = (StringUtils.isNotEmpty(param.getServerUrl())?param.getServerUrl():this.prop.getFlinkAddress()) + "jars/" 
                + param.getJarId()+"/run";
	    //如果成功，返回的内容形如:{"jobid":"7bc030d487a37c3b23b85de4eb2ad26f"}
	    if (StringUtils.isEmpty(param.getJobName())) {
	    	param.setJobName("lzf-test-"+String.valueOf((new Date()).getTime()));
	    }
	    try {
	        Map<String,Object> body=new HashMap<>();
	        body.put("entryClass","");
	        body.put("savepointPath",null);
	        body.put("parallelism",1);
	        body.put("allowNonRestoredState",null);
	        
	        //方法的的参数第一个是作业名称，第二个是artemis的密码，第三个mysql数据库的密码
	        if (param.getProgramArgs()==null || param.getProgramArgs().size()==0) {
	        	String[] args=new String[] {param.getJobName()};
	        	body.put("programArgsList",args);
	        }
	        else {
	        	List<String> argList=param.getProgramArgs();
	        	String[] args=new String[argList.size()+1];
	        	args[0]=param.getJobName();
	        	for(int i=1;i<args.length;i++) {
	        		args[i]=argList.get(i-1);
	        	}
	        	body.put("programArgsList",args);
	        }
	        WebClient webClient = WebClient.create();
	        String response = webClient.post()
	                .uri(url)
	                .bodyValue(body)
	                .retrieve()
	                .onStatus(status -> !status.is2xxSuccessful(),
	                        clientResponse -> Mono.error(new RuntimeException("Failed to cancel job: " + clientResponse.statusCode())))
	                .bodyToMono(String.class)
	                .block(); // 阻塞等待结果，保持与原有接口兼容
	        if (response!=null && response.contains("jobid"))
	            return PubResult.ok_data(response);
	        else
	            return PubResult.error("执行失败:" + response);
	    } catch (WebClientResponseException e) {
	        return PubResult.error("Error cancelling job: " + e.getStatusCode() + " - " + e.getMessage());
	    } catch (Exception e) {
	        return PubResult.error("Error connecting to Flink server: " + e.getMessage());
	    }
	}

	public PubResult<String> uploadJar(MultipartFile file,String serverUrl) {
        try {
            // 1. 保存文件到本地临时目录
            String tempDir = prop.getUploadDirectory();
            String fileName = file.getOriginalFilename();
            Path tempFilePath = Paths.get(tempDir, fileName);
            file.transferTo(tempFilePath);

            // 2. 使用WebClient上传到Flink集群
            WebClient webClient = WebClient.create();
            String url = (StringUtils.isNotEmpty(serverUrl)?serverUrl:this.prop.getFlinkAddress()) + "jars/upload";
            String response = webClient.post()
                .uri(url)
                .contentType(MediaType.MULTIPART_FORM_DATA)
                .body(BodyInserters.fromMultipartData("jarfile", new FileSystemResource(tempFilePath.toFile())))
                .retrieve()
                .onStatus(status -> !status.is2xxSuccessful(),
                    clientResponse -> Mono.error(new RuntimeException("Failed to upload jar: " + clientResponse.statusCode())))
                .bodyToMono(String.class)
                .block();

            // 3. 删除临时文件
            Files.deleteIfExists(tempFilePath);

            return PubResult.ok_data(response);
        } catch (WebClientResponseException e) {
            return PubResult.error("Error uploading jar: " + e.getStatusCode() + " - " + e.getMessage());
        } catch (IOException e) {
            return PubResult.error("Error handling file: " + e.getMessage());
        } catch (Exception e) {
            return PubResult.error("Error uploading jar: " + e.getMessage());
        }
    }


}
